#ifndef MQTT_PUSH_H_
#define MQTT_PUSH_H_

#include "common.h"
#include "json.h"
#include <thread>
#include <atomic>

extern "C"
{
#include "MQTTClient.h"
}

using json = nlohmann::json;

class MqttPush
{

public:
    MqttPush(std::string address, std::string topic, std::string mqtt_client_id, int max_que_size = 20)
        : mqtt_address_(address), mqtt_topic_(topic), mqtt_clientId(mqtt_client_id), stop_flag_(false), max_que_size_(max_que_size)
    {
        MQTTClient_create(&mqtt_cli, mqtt_address_.c_str(), mqtt_clientId.c_str(), MQTTCLIENT_PERSISTENCE_NONE, NULL);
        conn_opts.keepAliveInterval = 20;
        conn_opts.cleansession = 1;

        bm_dev_request(&handle_, 0);
        // workThread = std::thread(&MqttPush::dowork, this);
    }
    ~MqttPush()
    {
        stop_flag_ = true;
        workThread->join();
    };

    void start()
    {
        stop_flag_ = false;
        std::cout << "mqtt push: " + mqtt_address_ + mqtt_topic_ + " start!" << std::endl;
        workThread = std::make_shared<std::thread>(&MqttPush::dowork, this);
    }

    void stop()
    {
        stop_flag_ = true;
        std::cout << "mqtt push: " + mqtt_address_ + mqtt_topic_ + " stop!" << std::endl;
    }

    void add_channel(int channel_idx);
    void push_img(int channel_idx, std::shared_ptr<FrameInfoDetectYolov8Seg> img_ptr);

private:
    MQTTClient mqtt_cli;
    MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
    MQTTClient_message pubmsg = MQTTClient_message_initializer;
    MQTTClient_deliveryToken token;

    std::string mqtt_address_;
    std::string mqtt_clientId = "mqtt_client";
    std::string mqtt_username = "mqtt";
    std::string mqtt_password = "mqtt";
    std::string mqtt_topic_;
    int qos = 1;
    int timeout = 10000;

    bm_handle_t handle_;

    std::shared_ptr<std::thread> workThread;
    std::string path;

    std::atomic<bool> stop_flag_;
    std::mutex stop_flag_mtx_;

    int max_que_size_;
    std::deque<std::pair<int, std::shared_ptr<FrameInfoDetectYolov8Seg>>> img_que_;
    std::mutex img_que_mtx_;

    std::vector<int> channel_idx_vec_;
    std::vector<std::shared_ptr<std::thread>> thread_vec_;

    std::string base64_encode_bmcv(bm_handle_t handle_, unsigned char *jpegData, size_t nBytes);
    void dowork();

    bool get_stop_flag()
    {
        std::lock_guard<std::mutex> lock(stop_flag_mtx_);
        return stop_flag_;
    }
};

void MqttPush::add_channel(int channel_idx)
{
    channel_idx_vec_.push_back(channel_idx);
}

void MqttPush::push_img(int channel_idx, std::shared_ptr<FrameInfoDetectYolov8Seg> img_ptr)
{
    // if (channel_idx_vec_.empty())
    // {
    //     std::cout << "channel_idx is empty" << std::endl;
    //     return;
    // }
    // if (channel_idx_vec_.end() == std::find(channel_idx_vec_.begin(), channel_idx_vec_.end(), channel_idx))
    // {
    //     std::cout << "channel_idx not found" << std::endl;
    //     return;
    // }
    {
        std::lock_guard<std::mutex> lock(img_que_mtx_);
        if (img_que_.size() >= max_que_size_)
        {
            std::cout << "http_lib_push img_que is full" << std::endl;
            img_que_.pop_front();
        }
        img_que_.emplace_back(std::make_pair(channel_idx, img_ptr));
    }
}

std::string MqttPush::base64_encode_bmcv(bm_handle_t handle_, unsigned char *jpegData,
                                         size_t nBytes)
{
    // for bmcv
    unsigned long origin_len[2] = {nBytes, 0};
    unsigned long encode_len[2] = {(origin_len[0] + 2) / 3 * 4, 0};
    std::string res(encode_len[0], '\0');
    bmcv_base64_enc(handle_, bm_mem_from_system(jpegData),
                    bm_mem_from_system(const_cast<char *>(res.c_str())),
                    origin_len);
    return res;
}

void MqttPush::dowork()
{
    std::cout << "MQTT push doworking" << std::endl;
    int rc = 0;
    unsigned int reconnet_times = 0;
    int push_fail_times = 0;

    while (true)
    {
        std::cout << "mqtt reconnet stream times: " << reconnet_times++ << "  url:" + mqtt_address_ << std::endl;

        if (get_stop_flag())
        {
            break;
        }
        if ((rc = MQTTClient_connect(mqtt_cli, &conn_opts)) != MQTTCLIENT_SUCCESS)
        {
            std::cout << "Failed to connect, return code" << rc << std::endl;
            usleep(1000 * 1000);
            continue;
        }
        else
        {
            std::cout << "MQTT push connected" << std::endl;

            while (true)
            {
                if (get_stop_flag())
                {
                    break;
                }

                std::unique_lock<std::mutex> lock(img_que_mtx_);
                if (!img_que_.empty())
                {
                    std::shared_ptr<FrameInfoDetectYolov8Seg> img_ptr;
                    int channel_idx;
                    {
                        channel_idx = img_que_.front().first;
                        img_ptr = img_que_.front().second;
                        img_que_.pop_front();
                        lock.unlock();
                    }
                    {
                        nlohmann::json json_data_obj;
                        json_data_obj["channel_idx"] = channel_idx;
                        // json_data_obj["channel_url"] =
                        json_data_obj["is_mask"] = img_ptr.get()->is_mask;
                        json_data_obj["width"] = img_ptr.get()->image_ptr->width;
                        json_data_obj["height"] = img_ptr.get()->image_ptr->height;
#if 0
                        // jpeg 的 base64 编码耗资源，耗带宽，可暂时不用
                        std::string img_base64_data;
                        void *jpeg_data = NULL;
                        size_t out_size = 0;
                        bmcv_image_jpeg_enc(handle_, 1, img_ptr->image_ptr.get(),
                                            &jpeg_data, &out_size);
                        img_base64_data = base64_encode_bmcv(handle_,
                                                             (unsigned char *)jpeg_data, out_size);
                        free(jpeg_data);
                        json_data_obj["img_base64_data"] = img_base64_data;

                        // yolov8seg 检测结果
                        std::vector<YoloV8Box> boxs_vec = img_ptr->boxs_vec;
                        json box_json = json::array();
                        for (auto &box : boxs_vec)
                        {
                            json hullPoints_json = json::array();
                            for (auto &point : box.hullPoints)
                            {
                                hullPoints_json.push_back(
                                    json({{"x", point.x}, {"y", point.y}}));
                            }
                            box_json.push_back(
                                json(json::object({{"x1", box.x1},
                                                   {"y1", box.y1},
                                                   {"x2", box.x2},
                                                   {"y2", box.y2},
                                                   {"score", box.score},
                                                   {"class_id", box.class_id},
                                                   {"class_name", box.class_name},
                                                   //    {"mask", box.mask}, // 暂时不传mask数据，意义不大
                                                   {"angle", box.angle},
                                                   {"length", box.length},
                                                   {"hullPoints", hullPoints_json}})));
                        }
                        json_data_obj["boxs"] = box_json;
#endif

                        std::string json_data = json_data_obj.dump();
                        // std::cout << "json_data: " << json_data << std::endl;
                        pubmsg.payload = const_cast<char *>(json_data.c_str());
                        pubmsg.payloadlen = json_data_obj.dump().size();
                        pubmsg.qos = 2;
                        pubmsg.retained = 0;
                        MQTTClient_publishMessage(mqtt_cli, mqtt_topic_.c_str(), &pubmsg, &token);
                        // std::cout << "Waiting for up to" << (int)(timeout / 1000) << "seconds for publication of " << json_data_obj.dump() << "\n"
                        //           << "on topic" << mqtt_topic_ << "for client with ClientID: " << mqtt_clientId << "\n"
                        //           << std::endl;

                        rc = MQTTClient_waitForCompletion(mqtt_cli, token, 1000);
                        if (rc != MQTTCLIENT_SUCCESS)
                        {
                            std::cout << "Failed to publish message with token" << rc << std::endl;
                            push_fail_times++;
                        }
                        // std::cout << "Message with delivery token " << token << " delivered" << std::endl;
                    }
                }
                if (push_fail_times > 5)
                {
                    push_fail_times = 0;
                    break; // 断开连接，重新连接broker
                }
            }
            MQTTClient_disconnect(mqtt_cli, 10000);
            std::cout << "MQTTClient_disconnect " << std::endl;
        }
    }
}
#endif // HTTP_LIB_PUSH_H_